package com.atguigu.Flink.sql.query;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink09_TopN {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Schema schema = Schema.newBuilder()
                .column("user", "STRING")
                .column("url", "STRING")
                .column("ts", "BIGINT")
                //指定处理时间字段
                .columnByExpression("pt", "proctime()")
                //指定事件时间字段
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts, 3) ")
                .watermark("et", "et - INTERVAL '0' SECOND ")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        tableEnv.createTemporaryView("t1",table);

        //基于窗口汇总每个url的点击次数
        String windowsql =
                "select url,count(*) cnt,window_start,window_end " +
                        "from table(tumble(table t1,descriptor(et),interval '10' second))" +
                        " group by window_start,window_end,url";
        Table table1 = tableEnv.sqlQuery(windowsql);

        tableEnv.createTemporaryView("t2",table1);
        //2. 基于over， 通过点击次数求排名
        String oversql =
                " select url , cnt , window_start ,window_end , " +
                        " ROW_NUMBER() over(partition by window_start ,window_end  order by cnt desc ) rk " +
                        " from t2" ;
        Table table2 = tableEnv.sqlQuery(oversql);
        tableEnv.createTemporaryView("t3",table2);

        //基于排名，过滤求出top2
        String rankSql =
                " select url ,cnt , window_start ,window_end , rk " +
                        " from t3 " +
                        " where rk <= 2" ;
        tableEnv.sqlQuery(rankSql).execute().print();



        tableEnv.sqlQuery(oversql).execute().print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
